package com.haoxi.gateway.test.gatewayserver;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Date;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.haoxi.gateway.common.TestConfConst;
import com.haoxi.gateway.test.SGIPConstant;
import com.haoxi.gateway.test.message.SGIPActiveTestMessage;
import com.haoxi.gateway.test.message.SGIPActiveTestRepMessage;
import com.haoxi.gateway.test.message.SGIPBindMessage;
import com.haoxi.gateway.test.message.SGIPBindRepMessage;
import com.haoxi.gateway.test.message.SGIPDeliverMessage;
import com.haoxi.gateway.test.message.SGIPDeliverRepMessage;
import com.haoxi.gateway.test.message.SGIPReportMessage;
import com.haoxi.gateway.test.message.SGIPReportRepMessage;
import com.haoxi.gateway.test.util.TypeConvert;

/**
 * 
 * @author haoxi
 * 
 */
public class ReceiverSpThread extends Thread {
    protected static final Log log = LogFactory.getLog(ReceiverSpThread.class);
    private Socket socket = null;
    private OutputStream out = null;
    private InputStream in = null;
    protected boolean connectedState = false;
    private boolean bindState = false;
    public Hashtable reportData = new Hashtable();

    public void run() {
        // 与SP连接
        connectToSp();
        // 进行绑定
        parseBindOp();
        // 定时下发deliver消息
        // 定时下发report消息
        // 启动定时线程，向客户端发deliver消息
        Timer timer = new Timer(true);
        timer.schedule(new SendDeliverTask(), new Date(System
                .currentTimeMillis() + 10000), 10000);
        // Timer timer1 = new Timer(true);
        timer.schedule(new ReportToSpThread(), new Date(), 1000);
        timer.schedule(new HeardToSpThread(), new Date(), 30000);
        new ReceiverMsgThread().start();

        // 死循环检测连接是否正常
        while (true) {
            if (!connectedState) {
                // 与SP连接
                connectToSp();
            }
            if (!bindState) {
                // 进行绑定
                parseBindOp();
            }
        }

    }

    /**
     * 不断接收SP服务端发来的消息
     * 
     * @author haoxi
     * 
     */
    public class ReceiverMsgThread extends Thread {
        public void run() {
            try {
                if (socket == null || socket.isClosed()) {
                    return;
                }
                byte[] allbytes = null;
                byte[] headbytes = new byte[20];
                byte[] bodybytes = null;
                int commonLength = 0;
                while (true) {
                    in.read(headbytes);
                    commonLength = TypeConvert.byte2int(headbytes, 0);
                    bodybytes = new byte[commonLength - headbytes.length];
                    in.read(bodybytes);
                    allbytes = new byte[commonLength];
                    // 处理联通SGIP协议
                    System.arraycopy(headbytes, 0, allbytes, 0, 20);
                    System.arraycopy(bodybytes, 0, allbytes, 20,
                            bodybytes.length);
                    parseMsg(allbytes);

                }

            } catch (Exception e) {
                log.error("error !!! handling a client:", e);
            }
        }
    }

    public class ReportToSpThread extends TimerTask {
        public void run() {
            if (null == socket || socket.isClosed()) {
                return;
            }
            Set set = MOConnHandler.reportData.keySet();
            String key = "";
            // 复制一份临时的set，以免遍历时发生错误
            Set tmp = new HashSet();
            tmp.addAll(set);
            SGIPReportMessage report = null;
            Iterator iter = tmp.iterator();
            synchronized (iter) {
                while (iter.hasNext()) {
                    key = (String) iter.next();
                    try {
                        report = (SGIPReportMessage) MOConnHandler.reportData
                                .remove(key);
                        log.debug("---------send report message------");
                        out.write(report.getBytes());
                        out.flush();
                        log.debug(report.toString());
                        sleep(20);

                    } catch (IOException ex) {
                        log.error("exception in reportToSpThread.run()", ex);
                        connectedState = false;
                    } catch (InterruptedException ie) {
                        continue;
                    }
                }
                // 遍历结束清除掉临时的set
                tmp.clear();
            }

        }
    }

    public class HeardToSpThread extends TimerTask {
        public void run() {
            if (null == socket || socket.isClosed()) {
                return;
            }
            SGIPActiveTestMessage active = new SGIPActiveTestMessage();
            try {
                out.write(active.getBytes());
                out.flush();
                log.debug(active.toString());
            } catch (IOException eio) {
                log.error("exception in HeardToSpThread.run()", eio);
            }

        }
    }

    // 入参：待处理的消息
    // 返回：未处理完剩余的消息
    // public byte[] parseMsg(byte[] buf) {
    // int alllength = buf.length;
    // int msglenth = TypeConvert.byte2int(buf, 0);
    // if (0 == msglenth) {
    // log.debug("bad messages,message length is 0");
    // return new byte[0];
    // }
    // if (alllength < msglenth) {
    // return buf;
    // }
    // byte[] msgbytes = new byte[msglenth];
    // System.arraycopy(buf, 0, msgbytes, 0, msglenth);
    // // parse处理具体的业务
    // try {
    // int commid = TypeConvert.byte2int(msgbytes, 4);
    // log.debug("received sp message");
    // if (SGIPConstant.Deliver_Rep_Command_Id == commid) {
    // SGIPDeliverRepMessage deliverMessage = new SGIPDeliverRepMessage(
    // msgbytes);
    // log.debug(deliverMessage.toString());
    // // 结束
    // } else if (SGIPConstant.Report_Rep_Command_Id == commid) {
    // SGIPReportRepMessage deliverMessage = new SGIPReportRepMessage(
    // msgbytes);
    // log.debug(deliverMessage.toString());
    // // 结束
    // } else if (SGIPConstant.Bind_Rep_Command_Id == commid) {
    // SGIPBindRepMessage repMeg = new SGIPBindRepMessage(msgbytes);
    // if (repMeg.getResult() == 0) {
    // bindState = true;
    // }
    // log.debug(repMeg.toString());
    // } else if (SGIPConstant.ACTIVE_TEST_RESP == commid) {
    // SGIPActiveTestRepMessage activerep = new SGIPActiveTestRepMessage(
    // msgbytes);
    // log.debug(activerep.toString());
    // }
    // } catch (Exception ex) {
    // log.error("exception ", ex);
    // }
    // int spare = alllength - msglenth;
    // // 将剩余的全部拷贝到临时变量里面去
    // byte[] tmp = new byte[spare];
    // System.arraycopy(buf, msglenth, tmp, 0, spare);
    // // 不够截取一个消息头
    // if (4 > spare) {
    // return tmp;
    // }
    // msglenth = TypeConvert.byte2int(tmp, 0);
    // // 截取完后还有剩余的字节
    // if (msglenth < spare) {
    // return parseMsg(tmp);
    // }
    // // 不够一条消息
    // else {
    // return tmp;
    // }
    // }
    public void parseMsg(byte[] msgbytes) {
        try {
            int commid = TypeConvert.byte2int(msgbytes, 4);
            log.debug("received sp message");
            if (SGIPConstant.Deliver_Rep_Command_Id == commid) {
                SGIPDeliverRepMessage deliverMessage = new SGIPDeliverRepMessage(
                        msgbytes);
                log.debug(deliverMessage.toString());
                // 结束
            } else if (SGIPConstant.Report_Rep_Command_Id == commid) {
                SGIPReportRepMessage deliverMessage = new SGIPReportRepMessage(
                        msgbytes);
                log.debug(deliverMessage.toString());
                // 结束
            } else if (SGIPConstant.Bind_Rep_Command_Id == commid) {
                SGIPBindRepMessage repMeg = new SGIPBindRepMessage(msgbytes);
                if (repMeg.getResult() == 0) {
                    bindState = true;
                }
                log.debug(repMeg.toString());
            } else if (SGIPConstant.ACTIVE_TEST_RESP == commid) {
                SGIPActiveTestRepMessage activerep = new SGIPActiveTestRepMessage(
                        msgbytes);
                log.debug(activerep.toString());
            }
        } catch (Exception ex) {
            log.error("exception ", ex);
        }
    }

    public boolean parseBindOp() {
        if (null == socket || socket.isClosed()) {
            return false;
        }
        // byte[] readbytes = new byte[512];
        try {
            if (connectedState) {
                SGIPBindMessage bindMsg = new SGIPBindMessage(2,
                        TestConfConst.login_Name, TestConfConst.login_Passowrd);
                log.debug("---------send bind message------");
                out.write(bindMsg.getBytes());
                out.flush();
                log.debug(bindMsg.toString());
                try {
                    sleep(10000);
                } catch (InterruptedException ie) {
                    log.debug("sleep error", ie);
                }
            }

        } catch (IOException ex) {
            log.error("exception in reportToSpThread.parseBindOp()", ex);
            bindState = false;
            connectedState = false;
        }
        return bindState;

    }

    public boolean isConnection() {
        return connectedState;
    }

    public void disConnected() {
        this.connectedState = false;

        try {
            if (this.in != null) {
                this.in.close();
                Thread.sleep(200);
            }
        } catch (Exception e) {
            log.error("has a exception at input close:", e);
        } finally {
            this.in = null;
        }
        try {
            if (this.out != null) {
                this.out.close();
                Thread.sleep(200);
            }
        } catch (Exception e) {
            log.error("has a exception at output close:", e);
        } finally {
            this.out = null;
        }
        try {
            if (this.socket != null) {
                this.socket.close();
                Thread.sleep(200);
            }
        } catch (Exception e) {
            log.error("has a exception at socket close:", e);
        } finally {
            this.socket = null;
        }
    }

    public class SendDeliverTask extends TimerTask {
        public void run() {
            if (null == socket || socket.isClosed() || (!bindState)
                    || TestConfConst.deliverstate == 0) {
                return;
            }
            try {
                SGIPDeliverMessage deliver = new SGIPDeliverMessage(
                        TestConfConst.src_terminal_Id, TestConfConst.dest_Id,
                        1, 1, TestConfConst.msgFmt, TestConfConst.msg_Content
                                .getBytes(), TestConfConst.linkID);
                for (int i = 0; i < TestConfConst.delivertimes; i++) {
                    if (socket.isBound() && socket.isConnected()) {
                        deliver = new SGIPDeliverMessage(
                                TestConfConst.src_terminal_Id,
                                TestConfConst.dest_Id, 1, 1,
                                TestConfConst.msgFmt, TestConfConst.msg_Content
                                        .getBytes(), TestConfConst.linkID);
                        log.debug("---------send deliver message------");
                        out.write(deliver.getBytes());
                        out.flush();
                        log.debug(deliver.toString());
                        sleep(20);
                    }
                }

            } catch (IOException ioe) {
                log.error("ioexception", ioe);
            } catch (InterruptedException ie) {
                log.error("InterruptedException", ie);
            }
        }
    }

    public synchronized boolean connectToSp() {
        int connTimes = 0;
        while (!connectedState) // 当连接不能正常时，需重复继续连接，直到连接正常
        {
            try {
                // first dis connect
                this.disConnected();
                try {
                    connTimes++;
                    log.debug("host:" + TestConfConst.sphost + "  port:"
                            + TestConfConst.spport);
                    socket = new Socket(TestConfConst.sphost,
                            TestConfConst.spport);
                    socket.setSoTimeout(30000);
                    in = socket.getInputStream();
                    out = socket.getOutputStream();
                    // default socket connected
                    connectedState = true;
                    log.debug("connect ok!");
                    // if (connectedState) {
                    // break;
                    // }
                    if (connTimes >= 5) {
                        connTimes = 0;
                        // // if 15 times not connect, the program is exit
                        sleep(5000);
                    }
                } catch (Exception e) {
                    connectedState = false;
                    log.error("connect error", e);
                } finally {
                    if (!connectedState) {
                        // 如果连接不成功，sleep一段时间，继续连接
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException e) {
                            log.debug("sleep:", e);
                        }
                    }
                }
            } catch (Exception e) {
                connectedState = false;
                log.error("find a Exception:", e);
            }
        }

        return connectedState;

    }
    // in.read(readbytes);

}
